/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.messaging.rocket.producer;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.ipig.commons.codec.CodecService;
import org.ipig.commons.codec.kryo.GenericKryoCodec;
import org.ipig.commons.conf.rocket.producer.RocketProducerConf;
import org.ipig.constants.ResultStatus;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;

/**
 * 通用Producer服务实现
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id$$
 * @since 1.0
 */
@Slf4j
@NoArgsConstructor
public class GerericRocketProducer implements RocketProducerService {
    //编码
    @Setter
    @Getter
    @Resource(type = GenericKryoCodec.class)
    protected CodecService codecService;
    //生产者
    private DefaultMQProducer producer;
    //配置
    @Getter
    @Autowired(required = false)
    private RocketProducerConf conf;

    /**
     * @param conf RocketProducerConf
     */
    public GerericRocketProducer(RocketProducerConf conf) {
        super();
        this.conf = conf;
    }

    @Override
    public DefaultMQProducer getProducer() {
        if (producer == null) {
            try {
                init();
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }
        return producer;
    }

    @Override
    public ResultStatus send(String msg) {
        SendResult sendResult = null;
        try {
            Message message = new Message(getConf().getDufaultTopic(),
                    getConf().getDufaultTags(),
                    getConf().getDufaultKeys(), codecService.encode(msg));
            sendResult = getProducer().send(message);
        } catch (Exception e) {
            e.printStackTrace();
        }
        if (sendResult == null ||
                sendResult.getSendStatus() != SendStatus.SEND_OK) {
            return ResultStatus.FAILURE;
        } else {
            return ResultStatus.SUCCESS;
        }
    }

    /**
     * 初始化
     *
     * @throws MQClientException
     */
    public void init() throws MQClientException {
        if (producer == null) {
            synchronized (this) {
                if (producer == null) {
                    DefaultMQProducer mqProducer = new DefaultMQProducer();
                    mqProducer.setNamesrvAddr(conf.getNamesrvAddr());
                    mqProducer.setInstanceName(conf.getInstanceName());
                    mqProducer.setProducerGroup(conf.getGroup());
                    mqProducer.setDefaultTopicQueueNums(conf.getQueueNums());
                    mqProducer.setSendMsgTimeout(conf.getTimeout());
                    mqProducer.setMaxMessageSize(conf.getMaxMessageSize());
                    mqProducer.start();
                    producer = mqProducer;
                }
            }
        }
    }
}